package com.demo.service.impl;

import com.demo.service.KafkaService;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import java.util.Date;
import java.util.Properties;

@Service
public class KafkaServiceImpl implements KafkaService {

    private Logger log = LoggerFactory.getLogger(KafkaService.class);

/*
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
*/

    private String TOPIC = "con";

    KafkaProducer<String, String> producer;

    {
        System.out.println("===");

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "node2:9092,node3:9092,node4:9092"); //kafka连接地址
        properties.put("key.serializer", StringSerializer.class);
        properties.put("value.serializer", StringSerializer.class);

        //创建生产者
        producer = new KafkaProducer<String, String>(properties);

        System.out.println("===");
    }

    @Override
    public void send(String key, String value) {
        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC, key, value);

        producer.send(producerRecord);

       /* ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send(TOPIC, key, value);

        send.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

            @Override
            public void onFailure(Throwable throwable) {
                log.error("kafka send msg err, ex = {}, topic = {}, data = {}", throwable, TOPIC, value);
            }

            @Override
            public void onSuccess(SendResult<String, String> integerStringSendResult) {
                log.info("kafka send msg success, topic = {}, data = {}", TOPIC, value);
            }
        });*/
    }

    @Override
    public String makeLog(String userId, String productId, String action) {

        StringBuilder sb = new StringBuilder();
        sb.append(userId);
        sb.append(",");
        sb.append(productId);
        sb.append(",");
        sb.append(getSecondTimestamp(new Date()));
        sb.append(",");
        sb.append(action);
        return sb.toString();
    }

    public static String getSecondTimestamp(Date date) {
        if (null == date) {
            return "";
        }
        String timestamp = String.valueOf(date.getTime());
        int length = timestamp.length();
        if (length > 3) {
            return timestamp.substring(0, length - 3);
        } else {
            return "";
        }
    }
}
